/*
*消费者消费log日志
 */
package nsq

import (
	"time"

	"github.com/nsqio/go-nsq"
)

type ConsumerInterface interface {
	HandelMsg(msg []byte)
}

type Consumer struct {
	nsqdAddr string
	topic    string
	channel  string
	consumer *nsq.Consumer
}

func NewConsumer(nsqdAddr, topic, channel string) (*Consumer, error) {
	var err error
	consumer := &Consumer{nsqdAddr: nsqdAddr, topic: topic, channel: channel}
	cfg := nsq.NewConfig()
	//设置重连时间
	cfg.LookupdPollInterval = time.Second
	//cfg.MaxInFlight = 1
	cfg.OutputBufferTimeout = time.Millisecond * 50
	consumer.consumer, err = nsq.NewConsumer(topic, channel, cfg)
	//屏蔽系统日志
	consumer.consumer.SetLogger(nil, 0)
	if err != nil {
		return nil, err
	}
	return consumer, nil
}

// 消费者
func (self *Consumer) Start(consumer ConsumerInterface) error {
	// 设置消息处理函数
	self.consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		go consumer.HandelMsg(message.Body)
		return nil
	}))
	// 连接到单例nsqd
	// if err := self.consumer.ConnectToNSQD(self.nsqdAddr); err != nil {
	// 	return err
	// }
	if err := self.consumer.ConnectToNSQLookupd(self.nsqdAddr); err != nil {
		return err
	}
	<-self.consumer.StopChan
	return nil
}
